package core

/*
rtmp代理的核心程序，
1、首先接收播放端的连接请求，并通过play的请求来获取需要播放的视频流地址，
2、根据播放地址查询流信息服务器，获取流所在的服务器；
3、访问流服务器请求建立连接；
4、建立连接后，请求进行播放，并获取chunksize等信息，并返回给播放器；
5、建立双方网络通道，进行socket读写转发；
version:20220119
author：xiaobo
date:2022-01-19
*/

import (
	"fmt"
	"io/ioutil"
	"net"
	"net/http"
	"os"
	"strings"
	"time"

	"github.com/garyburd/redigo/redis"
	"github.com/rs/zerolog"
	"github.com/rs/zerolog/log"
)

var logger = zerolog.New(os.Stdout).With().Timestamp().Logger()

//开始进行TCP处理的协程，连接的conn是请求的socket，backend是源服务器的IP地址;
func ConnectionHandler(conn *Conn, call_url string, pool *redis.Pool, qtype int, ch chan int) (err error) {

	// buffer := make([]byte, 1024)
	defer conn.Close()
	defer func(ch chan int) {
		<-ch
	}(ch)
	//与客户端进行握手；
	log.Debug().Msg("接收到客户端的握手请求")
	if err = conn.HandshakeServer(); err != nil {
		return
	}
	//握手后，会收到客户端的connect命令；
	connServer := NewConnServer(conn)
	log.Debug().Msg("接收到客户端的connect请求")
	err = connServer.ReadMsg()
	if err != nil {
		conn.Close()
		log.Error().Msg("readMsg Error")
		return err
	}
	appname, streamname, _ := connServer.GetInfo()
	log.Debug().Str("appname", appname).Str("streamname", streamname).Msg("客户端请求播放地址")
	//如果appname和streamname为空，则退出;
	if appname == "" || streamname == "" {
		return nil
	}
	// //通过服务器的查询接口获取服务器的地址；
	log.Debug().Str("call_url", call_url)
	ip := ""

	if qtype == 0 {

		ip = getLiveServerIPFromHttpRes(call_url, appname, streamname)
		if len(ip) == 0 {
			log.Error().Str("app", appname).Str("stream", streamname).Msg("没有获取到直播服务器的地址,可能该流信息有误!")
			return nil
		}
	} else {

		ip = getLiveServerIPFromRedis(pool, appname, streamname)
		if len(ip) == 0 {
			log.Error().Str("app", appname).Str("stream", streamname).Msg("没有获取到直播服务器的地址,可能该流信息有误!")
			return nil
		}
	}
	time.Sleep(time.Duration(50) * time.Millisecond)
	log.Debug().Str("app", appname).Str("stream", streamname).Str("server", ip).Msg("取直播服务器地址")
	//根据请求地址，向服务器端发起连接请求
	log.Debug().Str("server", ip).Msg("向服务器发起新连接")
	new_rtmp := fmt.Sprintf("rtmp://%s/%s/%s", ip, appname, streamname)

	connClient := NewConnClient()
	err = connClient.Start(new_rtmp, "play")
	if err != nil {
		log.Error().Err(err).Msg("向服务器发起播放请求错误1次")
		connClient.conn.Close()
		time.Sleep(time.Duration(200) * time.Millisecond)
		err = connClient.Start(new_rtmp, "play")
		if err != nil {
			log.Error().Err(err).Msg("向服务器发起播放请求错误2次")
			connClient.conn.Close()
			time.Sleep(time.Duration(200) * time.Millisecond)
			err = connClient.Start(new_rtmp, "play")
			if err != nil {
				log.Error().Err(err).Msg("向服务器发起播放请求错误3次")
				connClient.conn.Close()
				return err
			}
		}
	}
	log.Debug().Int32("chunk_size", int32(connClient.conn.remoteChunkSize)).Int32("windowAckSize", int32(connClient.conn.remoteWindowAckSize)).Msg("connect调用后")

	reSendControlMsg(connServer, connClient.conn.remoteChunkSize, connClient.conn.remoteWindowAckSize, connClient.conn.remoteWindowAckSize)
	//握手完成后，则进入到proxy的处理过程；
	Proxy(conn.Conn, connClient.conn.Conn)
	return nil
}

func reSendControlMsg(connServer *ConnServer, chunksize, acknowlegment, bandwidth uint32) error {

	err := connServer.NewSetChunkSize(chunksize)
	if err != nil {
		return err
	}
	err = connServer.NewSetAcknowlegment(acknowlegment)
	if err != nil {
		return err
	}
	err = connServer.NewSetBandwidth(bandwidth)
	if err != nil {
		return err
	}
	return nil
}

func Proxy(from net.Conn, to net.Conn) {

	defer to.Close()
	closed := make(chan bool, 2)
	go Proxy2(from, to, closed)
	go Proxy2(to, from, closed)
	<-closed
	log.Debug().Msg("协程网络连接已断开")
}

func Proxy2(from net.Conn, to net.Conn, closed chan bool) {

	buffer := make([]byte, 1024)
	for {
		n1, err := from.Read(buffer)
		if err != nil {
			logger.Error().Str("from", from.RemoteAddr().String()).Int("recv", n1).Str("to", to.RemoteAddr().String()).Err(err).Msg("与服务器网络断开")
			closed <- true
			return
		}
		// fmt.Println(3333, string(buffer[:n1]))
		n2, err := to.Write(buffer[:n1])
		// logger.Debug().Str("from", from.RemoteAddr().String()).Int("recv", n1).Str("to", to.RemoteAddr().String()).Int("send", n2).Send()
		if err != nil {
			closed <- true
			logger.Error().Str("from", from.RemoteAddr().String()).Int("send", n1).Str("to", to.RemoteAddr().String()).Int("send", n2).Err(err).Msg("与客户端网络断开")
			return
		}
	}
}

//通过http接口获取直播接口服务器的IP地址
func getLiveServerIPFromHttpRes(call_url, app, stream string) string {

	client := &http.Client{}

	//生成要访问的url,token是api鉴权，每个api访问方式不同，根据api调用文档拼接URL
	url := fmt.Sprintf("%s?app=%s&stream=%s", call_url, app, stream)
	//提交请求
	request, err := http.NewRequest("GET", url, nil)
	//异常捕捉
	if err != nil {
		panic(err)
	}

	//处理返回结果
	response, err := client.Do(request)
	if err != nil {
		log.Fatal().Err(err)
		return ""
	}
	defer response.Body.Close()
	//检出结果集
	ip, err := ioutil.ReadAll(response.Body)
	if err != nil {
		log.Fatal().Err(err)
		return ""
	}
	return strings.Replace(string(ip), "\n", "", -1)
}

//通过访问redis获取直播接口服务器的IP地址
func getLiveServerIPFromRedis(pool *redis.Pool, app, stream string) string {

	conn1 := pool.Get()
	defer conn1.Close()

	rkey := fmt.Sprintf("%s_%s_rtmp", app, stream)
	rvalue, err := redis.String(conn1.Do("Get", rkey))
	if err != nil {
		return ""
	}
	n := strings.Index(rvalue, "/")
	if n < 1 {
		return ""
	}
	str := rvalue[:n]
	s := strings.Split(str, ":")
	if len(s) < 2 {
		return ""
	}
	return s[0]
}
